/******************************************************************************
**
**                      (C)2023 Ninebot Limited
**
**                         All Right Reserved
**
** 1 tab == 4 spaces!
******************************************************************************/
#define LOG_TAG "j1939.tag"

#include <stdio.h>
#include <stdint.h>
#include <string.h>
#include <FreeRTOS.h>
#include <queue.h>
#include <task.h>
#include <timers.h>
#include "j1939tp.h"
#include "j1939tp_config.h"
#include "../drivers/drv_can.h"

#include "rtdef.h"

#include "elog.h"

#define T1_TIMEOUT 							1
#define T2_TIMEOUT 							2
#define T3_TIMEOUT 							3
#define T4_TIMEOUT 							4
#define SEND_CTS_EVENT 						5
#define SEND_EOMA_EVENT 					6
#define SEND_ABORT_EVENT 					7
#define SEND_DT_EVENT 						8

#define NO_LIMIT_PACKET_NUM 				0xFF

#define J1939_CTS_MSG 						0x01
#define J1939_EOMA_MSG 						0x02
#define J1939_ABORT_MSG 					0x03
#define J1939_MULTIFRAME_MSG 				0x04

#define RETRANSMIT_NO_LIMIT 				0xFF
#define HOLD_CONNECTION_OPEN_NUM 			0
#define NO_NEED_PORCESS_COMPLETE_FRAME 		-1

#define SPECIFIC_SESSION 					0
#define GLOBAL_SESSION 						1

/* 这个函数不能用于中断调用 */
void on_j1939tp_received(const int channel, const struct can_frame *cf, void *arg)
{
	j1939tp_runtime_t *runtime = (j1939tp_runtime_t *)arg;
	if (runtime->config->rxqueue != NULL)
	{
		{
			/* 使用非阻塞的方式，传入队列中 */
			if (!xQueueSend(runtime->config->rxqueue, cf, (TickType_t)0))
			{
				log_d("cantp layer receive can frame(chn:%d,id:%x) failed, may be queue is full now.\n", channel,
					  cf->can_id);
			}
		}
	}
}

/* 用于中断内部调用 */
void on_j1939tp_received_from_isr(const int channel, const struct can_frame *cf, void *arg)
{
    BaseType_t xHigherPriorityTaskWoken;
	j1939tp_runtime_t *runtime = (j1939tp_runtime_t *)arg;
	if (runtime->config->rxqueue != NULL)
	{
		{
			/* 使用非阻塞的方式，传入队列中 */
			if (!xQueueSendFromISR(runtime->config->rxqueue, cf, &xHigherPriorityTaskWoken))
			{
				log_d("cantp layer receive can frame(chn:%d,id:%x) failed, may be queue is full now.\n", channel,
					  cf->can_id);
			}
            portYIELD_FROM_ISR(xHigherPriorityTaskWoken);//如果需要的话进行一次任务切换
		}
	}
}

/* j1939tp输出的错误信息 */
static const char *j1939_tp_abort_to_str(enum j1939_tp_abort abort)
{
	switch (abort)
	{
	case J1939_TP_ABORT_BUSY:
		return "Already in one or more connection managed sessions and cannot support another.";
	case J1939_TP_ABORT_RESOURCE:
		return "System resources were needed for another task so this connection managed session was terminated.";
	case J1939_TP_ABORT_TIMEOUT:
		return "A timeout occurred and this is the connection abort to close the session.";
	case J1939_TP_ABORT_GENERIC:
		return "CTS messages received when data transfer is in progress";
	case J1939_TP_ABORT_FAULT:
		return "Maximal retransmit request limit reached";
	case J1939_TP_ABORT_UNEXPECTED_DATA:
		return "Unexpected data transfer packet";
	case J1939_TP_ABORT_BAD_SEQ:
		return "Bad sequence number (and software is not able to recover)";
	case J1939_TP_ABORT_DUP_SEQ:
		return "Duplicate sequence number (and software is not able to recover)";
	case J1939_TP_ABORT_MESSAGE_SIZE_EXCEED_LIMIT:
		return "Message size > 1785 bytes (TP)";
	case J1939_TP_ABORT_OTHER:
		return "Any other reason (if a Connection Abort reason is identified that is not listed in the table use code 250)";
	default:
		return "<unknown>";
	}
}

static int rx_ecu_bin_search(const j1939tp_rx_sesssion_t *tab, const uint8_t expect, const uint16_t len)
{
	uint16_t index_l = 0;
	uint16_t index_r = len - 1;
	uint16_t index_m = 0;

	while (index_l <= index_r)
	{
		index_m = index_l + ((index_r - index_l) >> 1); /*	calculate the middle index */

		if (tab[index_m].addr == expect)
			return index_m; /* already be found ,return middle index now */
		else if (tab[index_m].addr > expect)
			index_r = index_m - 1; /* search left side */
		else
			index_l = index_m + 1; /* search right side */
	}
	return -1; /* can't find any id */
}

static int tx_ecu_bin_search(const j1939tp_tx_sesssion_t *tab, const uint8_t expect, const uint16_t len)
{
	uint16_t index_l = 0;
	uint16_t index_r = len - 1;
	uint16_t index_m = 0;

	while (index_l <= index_r)
	{
		index_m = index_l + ((index_r - index_l) >> 1); /*	calculate the middle index */
        // log_d("index_r %d  index_l %d index_m %d", index_r, index_l, index_m);
		if (tab[index_m].addr == expect)
			return index_m; /* already be found ,return middle index now */
		else if (tab[index_m].addr > expect)
			index_r = index_m - 1; /* search left side */
		else
			index_l = index_m + 1; /* search right side */
	}
	return -1; /* can't find any id */
}

static int pgn_bin_search(const j1939_pgn_table_t *tab, const uint32_t expect, const uint16_t len)
{
	uint16_t index_l = 0;
	uint16_t index_r = len - 1;
	uint16_t index_m = 0;
    // log_d("len: %d", len);
	while (index_l <= index_r)
	{
		index_m = index_l + ((index_r - index_l) >> 1); /*	calculate the middle index */
        // log_d("index_m: %d, index_r: %d, index_l: %d", index_m, index_r, index_l);
		if (tab[index_m].pgn == expect)
			return index_m; /* already be found ,return middle index now */
		else if (tab[index_m].pgn > expect)
			index_r = index_m - 1; /* search left side */
		else
			index_l = index_m + 1; /* search right side */
	}
	return -1; /* can't find any id */
}

static inline bool j1939_pgn_is_pdu1(pgn_t pgn)
{
	/* ignore dp & res bits for this */
	return (pgn & 0xff00) < 0xf000;
}

static inline bool j1939_address_is_unicast(uint8_t addr)
{
	return addr <= J1939_MAX_UNICAST_ADDR;
}

static inline uint32_t j1939_get_canid(uint8_t saddr, uint8_t daddr, pgn_t pgn, priority_t priority)
{
	uint32_t id = 0;
	if (j1939_pgn_is_pdu1(pgn))
		id = (uint32_t)saddr | ((uint32_t)daddr << 8) | (pgn << 8) | ((uint32_t)priority << 26) | CAN_EFF_FLAG;
	else
		id = (uint32_t)saddr | (pgn << 8) | ((uint32_t)priority << 26) | CAN_EFF_FLAG;
	return id;
}

int j1939tp_write_by_channel(unsigned int channel, struct can_frame *cf)
{
	int ret = RT_EOK;
	configASSERT(cf != NULL);
	//	if (get_can_fd(channel) >= 0)
	//	{
	//		cf->sync_send_timeout = 0;
	//		cf->send_task_handle = xTaskGetCurrentTaskHandle();
	//		// return write(get_can_fd(channel), cf, sizeof(struct can_frame)) == sizeof(struct can_frame) ? 0 : -EBUSY;
	//	}
	//	else
	//		return -EIO;
	stm32_fdcan1_send_message(CAN_CHN1, cf->can_id, cf->data, cf->len);
	// log_d("can data send ch = %d", channel);
	return ret;
}

static void j1939tp_send_CTS_message(j1939tp_runtime_t *runtime, j1939tp_rx_sesssion_t *session)
{
	struct can_frame element;
	uint8_t len = 0;
	memset(&element, 0, sizeof(struct can_frame));
	memset(element.data, runtime->config->padding, 8);
	element.can_id =
		j1939_get_canid(session->runtime[0].saddr, session->addr, J1939TP_CM_PGN, session->runtime[0].priority);
	element.len = 8;
	element.data[0] = J1939_TP_CMD_CTS;

	len = SIZE2PACKETS(session->runtime[0].total_msg_size) - session->runtime[0].rx_total_packets;
	if (session->runtime[0].rx_max_packets == NO_LIMIT_PACKET_NUM)
		session->runtime[0].sendCTS_packets = min(len, (runtime->config->tp_block) ? (runtime->config->tp_block) : 255);
	else
	{
		len = min(len, (runtime->config->tp_block) ? (runtime->config->tp_block) : 255);
		session->runtime[0].sendCTS_packets = min(len, session->runtime[0].rx_max_packets);
	}
	element.data[1] = session->runtime[0].sendCTS_packets;
	element.data[2] = session->runtime[0].rx_total_packets + 1;

	element.data[5] = session->runtime[0].pgn & 0xFF;
	element.data[6] = (session->runtime[0].pgn & 0xFF00) >> 8;
	element.data[7] = (session->runtime[0].pgn & 0xFF0000) >> 16;

	j1939tp_write_by_channel(runtime->config->chn, &element);
}

static void j1939tp_send_ConnectionAbort_message(j1939tp_runtime_t *runtime, uint8_t sa, uint8_t da, uint32_t pgn,
												 uint8_t priority, uint8_t reason)
{
	struct can_frame element;
	memset(&element, 0, sizeof(struct can_frame));
	memset(element.data, runtime->config->padding, 8);
	element.can_id = j1939_get_canid(sa, da, J1939TP_CM_PGN, priority);
	element.len = 8;
	element.data[0] = J1939_TP_CMD_ABORT;
	element.data[1] = reason;

	element.data[5] = pgn & 0xFF;
	element.data[6] = (pgn & 0xFF00) >> 8;
	element.data[7] = (pgn & 0xFF0000) >> 16;

	j1939tp_write_by_channel(runtime->config->chn, &element);
}
static void j1939tp_send_EOMA_message(j1939tp_runtime_t *runtime, j1939tp_rx_sesssion_t *session)
{
	struct can_frame element;
	memset(&element, 0, sizeof(struct can_frame));
	memset(element.data, runtime->config->padding, 8);

	element.can_id =
		j1939_get_canid(session->runtime[0].saddr, session->addr, J1939TP_CM_PGN, session->runtime[0].priority);
	element.len = 8;
	element.data[0] = J1939_TP_CMD_EOMA;
	element.data[1] = session->runtime[0].total_msg_size & 0xFF;
	element.data[2] = (session->runtime[0].total_msg_size & 0xFF00) >> 8;
	element.data[3] = SIZE2PACKETS(session->runtime[0].total_msg_size);

	element.data[5] = session->runtime[0].pgn & 0xFF;
	element.data[6] = (session->runtime[0].pgn & 0xFF00) >> 8;
	element.data[7] = (session->runtime[0].pgn & 0xFF0000) >> 16;

	j1939tp_write_by_channel(runtime->config->chn, &element);
}
static inline void j1939tp_set_timer(j1939tp_internal_timer_t *timer, uint32_t timeout_ms, uint8_t event)
{
	timer->end_tick = xTaskGetTickCount() + pdMS_TO_TICKS(timeout_ms);
	timer->valid = 1;
	timer->timeout_type = event;
}
static inline void j1939tp_clear_timer(j1939tp_internal_timer_t *timer)
{
	timer->valid = 0;
	timer->end_tick = 0;
	timer->timeout_type = 0;
}
static inline void j1939tp_close_rx_session(j1939tp_rx_sesssion_t *session, uint8_t type)
{
	if (type <= GLOBAL_SESSION)
	{
		session->runtime[type].state = J1939TP_RX_IDLE;
		session->runtime[type].total_msg_size = 0;
		session->runtime[type].rx_total_packets = 0;
		session->runtime[type].sendCTS_packets = 0;
		session->runtime[type].rx_max_packets = 0;
		session->runtime[type].pgn = 0;
		session->runtime[type].saddr = 0;
		j1939tp_clear_timer(&session->runtime[type].timer);
	}
}
static inline void j1939tp_close_tx_session(j1939tp_tx_sesssion_t *session, uint8_t type)
{
	if (type <= GLOBAL_SESSION)
	{
		session->runtime[type].state = J1939TP_TX_IDLE;
		session->runtime[type].total_msg_size = 0;
		session->runtime[type].tx_total_packets = 0;
		session->runtime[type].waitForCTS_packets = 0;
		session->runtime[type].tx_max_packets = 0;
		session->runtime[type].retransmit_cnt = 0;
		session->runtime[type].pgn = 0;
		session->runtime[type].daddr = 0;
		j1939tp_clear_timer(&session->runtime[type].timer);
	}
}
static void j1939tp_cm_cmd_recv(j1939tp_runtime_t *runtime, j1939_msg_info_t *msg_info, struct can_frame *element)
{
	uint8_t cmd = element->data[0];
	int index = 0;
	j1939_output_msg_info_t msg = {0};
	pgn_t pgn = 0;
	switch (cmd)
	{
	case J1939_TP_CMD_RTS:
		if (msg_info->addr.da == J1939_BROADCAST_ADDR)
			break;
		pgn = (uint32_t)element->data[5] | (uint32_t)element->data[6] << 8 | (uint32_t)element->data[7] << 16;
		index = rx_ecu_bin_search(runtime->session->rx, msg_info->addr.sa, runtime->session->rx_num);
		if (index >= 0)
		{
			j1939tp_rx_sesssion_t *rx_session = &runtime->session->rx[index];
			if (rx_session->runtime[0].state == J1939TP_RX_IDLE || pgn == rx_session->runtime[0].pgn)
			{
				rx_session->runtime[0].pgn = pgn;
				rx_session->runtime[0].priority = msg_info->priority;
				rx_session->runtime[0].saddr = msg_info->addr.da;
				rx_session->runtime[0].total_msg_size = (uint32_t)element->data[1] | (uint32_t)element->data[2] << 8;
				if (rx_session->runtime[0].total_msg_size > 1785)
				{
					uint8_t sa = rx_session->runtime[0].saddr;
					uint8_t da = rx_session->addr;
					uint8_t priority = rx_session->runtime[0].priority;
					j1939tp_send_ConnectionAbort_message(runtime, sa, da, pgn, priority,
														 J1939_TP_ABORT_MESSAGE_SIZE_EXCEED_LIMIT);
					log_d("j1939tp:%s\n", j1939_tp_abort_to_str(J1939_TP_ABORT_MESSAGE_SIZE_EXCEED_LIMIT));
					j1939tp_close_rx_session(rx_session, SPECIFIC_SESSION);
				}
				else if (element->data[3] != SIZE2PACKETS(rx_session->runtime[0].total_msg_size))	/* 判断发过来的数据总长度和分包数量是否对应 */
				{
					uint8_t sa = rx_session->runtime[0].saddr;
					uint8_t da = rx_session->addr;
					uint8_t priority = rx_session->runtime[0].priority;
					j1939tp_send_ConnectionAbort_message(runtime, sa, da, pgn, priority,
														 J1939_TP_ABORT_UNEXPECTED_DATA);
					log_d("j1939tp:total size don't match total packets\n");
					j1939tp_close_rx_session(rx_session, SPECIFIC_SESSION);
				}
				else if (rx_session->runtime[0].total_msg_size > rx_session->max_len)
				{
					uint8_t sa = rx_session->runtime[0].saddr;
					uint8_t da = rx_session->addr;
					uint8_t priority = rx_session->runtime[0].priority;
					j1939tp_send_ConnectionAbort_message(runtime, sa, da, pgn, priority, J1939_TP_ABORT_RESOURCE);
					log_d("j1939tp:%s\n", j1939_tp_abort_to_str(J1939_TP_ABORT_RESOURCE));
					j1939tp_close_rx_session(rx_session, SPECIFIC_SESSION);
				}
				else
				{
					rx_session->runtime[0].state = J1939TP_RX_WAIT_SEND_CTS;
					rx_session->runtime[0].sendCTS_packets = 0;
					rx_session->runtime[0].rx_max_packets = element->data[4];
					rx_session->runtime[0].rx_total_packets = 0;
					j1939tp_set_timer(&rx_session->runtime[0].timer, 50, SEND_CTS_EVENT);
				}
			}
			else
			{
				uint8_t sa = rx_session->runtime[0].saddr;
				uint8_t da = rx_session->addr;
				uint8_t priority = rx_session->runtime[0].priority;
				j1939tp_send_ConnectionAbort_message(runtime, sa, da, pgn, priority, J1939_TP_ABORT_BUSY);
				log_d("j1939tp:%s\n", j1939_tp_abort_to_str(J1939_TP_ABORT_BUSY));
				j1939tp_close_rx_session(rx_session, SPECIFIC_SESSION);
			}
		}
		break;

	case J1939_TP_CMD_CTS:
	{
		/* 如果是广播地址直接跳出，不执行下面代码，这个是针对于点对点通讯的部分 */
		if (msg_info->addr.da == J1939_BROADCAST_ADDR)
			break;
		msg.msg_code = J1939_CTS_MSG;
		memcpy(&msg.frame, element, sizeof(struct can_frame));
		xQueueSend(runtime->config->txqueue, &msg, 0);
	}
	break;

	case J1939_TP_CMD_EOMA:
	{
		if (msg_info->addr.da == J1939_BROADCAST_ADDR)
			break;
		msg.msg_code = J1939_EOMA_MSG;
		memcpy(&msg.frame, element, sizeof(struct can_frame));
		xQueueSend(runtime->config->txqueue, &msg, 0);
	}
	break;

	case J1939_TP_CMD_BAM:
		if (msg_info->addr.da != J1939_BROADCAST_ADDR)
			break;
		pgn = (uint32_t)element->data[5] | (uint32_t)element->data[6] << 8 | (uint32_t)element->data[7] << 16;
		index = rx_ecu_bin_search(runtime->session->rx, msg_info->addr.sa, runtime->session->rx_num);
		if (index >= 0)
		{
			j1939tp_rx_sesssion_t *rx_session = &runtime->session->rx[index];
			if (rx_session->runtime[1].state == J1939TP_RX_IDLE || pgn == rx_session->runtime[1].pgn)
			{
				rx_session->runtime[1].pgn = pgn;
				rx_session->runtime[1].total_msg_size = (uint32_t)element->data[1] | (uint32_t)element->data[2] << 8;
				if (rx_session->runtime[1].total_msg_size <= rx_session->max_bam_len &&
					rx_session->runtime[1].total_msg_size <= 1785)
				{
					rx_session->runtime[1].state = J1939TP_RX_RECEIVING_DT;
					j1939tp_set_timer(&rx_session->runtime[1].timer, runtime->config->timeout_T1, T1_TIMEOUT);
				}
				else
				{
					j1939tp_close_rx_session(rx_session, GLOBAL_SESSION);
				}
				rx_session->runtime[1].rx_total_packets = 0;
			}
		}
		break;

	case J1939_TP_CMD_ABORT:
		if (msg_info->addr.da == J1939_BROADCAST_ADDR)
			break;
		pgn = (uint32_t)element->data[5] | (uint32_t)element->data[6] << 8 | (uint32_t)element->data[7] << 16;
		index = rx_ecu_bin_search(runtime->session->rx, msg_info->addr.sa, runtime->session->rx_num);
		if (index >= 0)
		{
			j1939tp_rx_sesssion_t *rx_session = &runtime->session->rx[index];
			if (rx_session->runtime[0].state != J1939TP_RX_IDLE && pgn == rx_session->runtime[0].pgn)
			{
				uint8_t reason = element->data[1];
				j1939tp_close_rx_session(rx_session, SPECIFIC_SESSION);
				log_d("j1939tp receive ABORT frame:%s", j1939_tp_abort_to_str(reason));
			}
		}
		msg.msg_code = J1939_ABORT_MSG;
		memcpy(&msg.frame, element, sizeof(struct can_frame));
		xQueueSend(runtime->config->txqueue, &msg, 0);
		break;
	}
}
int j1939tp_recv(j1939tp_runtime_t *runtime, j1939_msg_info_t *msg_info, struct can_frame *element)
{
	int index = 0;
	int ret = NO_NEED_PORCESS_COMPLETE_FRAME;
	switch (msg_info->addr.pgn)
	{
	case J1939TP_CM_PGN:
		j1939tp_cm_cmd_recv(runtime, msg_info, element);
		break;
	case J1939TP_DT_PGN:
		index = rx_ecu_bin_search(runtime->session->rx, msg_info->addr.sa, runtime->session->rx_num);
		if (index >= 0)
		{
			j1939tp_rx_sesssion_t *rx_session = &runtime->session->rx[index];
			if (j1939_address_is_unicast(msg_info->addr.da))
			{
				j1939tp_clear_timer(&rx_session->runtime[0].timer);
				if (rx_session->runtime[0].state == J1939TP_RX_RECEIVING_DT)
				{
					if ((rx_session->runtime[0].rx_total_packets + 1) <=
						SIZE2PACKETS(rx_session->runtime[0].total_msg_size))
					{
						if (element->data[0] == (rx_session->runtime[0].rx_total_packets + 1))	/* 下一包需要接收的包号 */
						{
							uint32_t pos = rx_session->runtime[0].rx_total_packets * 7;
							uint32_t len =
								rx_session->runtime[0].total_msg_size - rx_session->runtime[0].rx_total_packets * 7;
							if (len < 7)
								memcpy((rx_session->buffer + pos), &element->data[1], len);
							else
								memcpy((rx_session->buffer + pos), &element->data[1], 7);
							rx_session->runtime[0].rx_total_packets++;
							rx_session->runtime[0].sendCTS_packets--;
							if (rx_session->runtime[0].rx_total_packets >=
								SIZE2PACKETS(rx_session->runtime[0].total_msg_size))
							{
								rx_session->runtime[0].state = J1939TP_RX_WAIT_SEND_ENDOFMSGACK;
								ret = index;
								j1939tp_set_timer(&rx_session->runtime[0].timer, 20, SEND_EOMA_EVENT);
							}
							else if (rx_session->runtime[0].sendCTS_packets == 0)
							{
								rx_session->runtime[0].state = J1939TP_RX_WAIT_SEND_CTS;
								j1939tp_set_timer(&rx_session->runtime[0].timer, 50, SEND_CTS_EVENT);
							}
							else
							{
								j1939tp_set_timer(&rx_session->runtime[0].timer, runtime->config->timeout_T1,	/* 设置接收DT数据包的超时时间 */
												  T1_TIMEOUT);
							}
						}
						else if (element->data[0] == rx_session->runtime[0].rx_total_packets)	/* 重复的序列号 */
						{
							uint8_t sa = rx_session->runtime[0].saddr;
							uint8_t da = rx_session->addr;
							uint8_t priority = rx_session->runtime[0].priority;
							uint32_t pgn = rx_session->runtime[0].pgn;
							j1939tp_send_ConnectionAbort_message(runtime, sa, da, pgn, priority,
																 J1939_TP_ABORT_DUP_SEQ);
							log_d("j1939tp:%s\n", j1939_tp_abort_to_str(J1939_TP_ABORT_DUP_SEQ));
							j1939tp_close_rx_session(rx_session, SPECIFIC_SESSION);
						}
						else
						{
							uint8_t sa = rx_session->runtime[0].saddr;
							uint8_t da = rx_session->addr;
							uint8_t priority = rx_session->runtime[0].priority;
							uint32_t pgn = rx_session->runtime[0].pgn;
							j1939tp_send_ConnectionAbort_message(runtime, sa, da, pgn, priority,
																 J1939_TP_ABORT_BAD_SEQ);
							log_d("j1939tp:%s\n", j1939_tp_abort_to_str(J1939_TP_ABORT_BAD_SEQ));
							j1939tp_close_rx_session(rx_session, SPECIFIC_SESSION);
						}
					}
				}
			}
			else
			{
				j1939tp_clear_timer(&rx_session->runtime[1].timer);
				if (rx_session->runtime[1].state == J1939TP_RX_RECEIVING_DT)
				{
					if ((rx_session->runtime[1].rx_total_packets + 1) <=
						SIZE2PACKETS(rx_session->runtime[1].total_msg_size))
					{
						if (element->data[0] == (rx_session->runtime[1].rx_total_packets + 1))	/* 下一包需要接收的包号 */
						{
							uint32_t pos = rx_session->runtime[1].rx_total_packets * 7 + rx_session->max_len;	/* @这段的理解为跳跃点对点通讯的存储buffer长度 */
							uint32_t len =
								rx_session->runtime[1].total_msg_size - rx_session->runtime[1].rx_total_packets * 7;
							if (len < 7)
								memcpy((rx_session->buffer + pos), &element->data[1], len);
							else
								memcpy((rx_session->buffer + pos), &element->data[1], 7);
							rx_session->runtime[1].rx_total_packets++;
							if (rx_session->runtime[1].rx_total_packets >=
								SIZE2PACKETS(rx_session->runtime[1].total_msg_size))
							{
								rx_session->runtime[1].state = J1939TP_RX_IDLE;/*  */
								ret = index;
							}
							else
								j1939tp_set_timer(&rx_session->runtime[1].timer, runtime->config->timeout_T1,
												  T1_TIMEOUT);
						}
						else
						{
							uint8_t reason = 0;
							if (element->data[0] == rx_session->runtime[1].rx_total_packets)
								reason = J1939_TP_ABORT_DUP_SEQ;
							else
								reason = J1939_TP_ABORT_BAD_SEQ;
							log_d("j1939tp:%s", j1939_tp_abort_to_str(reason));
							j1939tp_close_rx_session(rx_session, GLOBAL_SESSION);
						}
					}
				}
			}
		}
		break;
	default:
		msg_info->addr.type = J1939_SIMPLE;
		ret = 0;
		break;
	}
	return ret;
}
void j1939tp_rx_task(void *arg)
{
	j1939tp_runtime_t *j1939tp_runtime = (j1939tp_runtime_t *)arg;
	const j1939tp_config_t *config = j1939tp_runtime->config;
	j1939tp_rx_sesssion_t *rx_session = j1939tp_runtime->session->rx;
	uint16_t rx_session_num = j1939tp_runtime->session->rx_num;
	TickType_t xNextExpireTime = 0;
	TickType_t xTimeNow = 0;
	uint16_t i = 0;
	j1939_msg_info_t j1939_msg_info;
	int complete_frame = 0;
	int ret = 0;
	struct can_frame element;
	memset(&element, 0, sizeof(struct can_frame));
	if (config->rxqueue)
	{
		xQueueCreateStatic(config->rx_queue_size, CAN_MTU, config->rxq_storage, config->rxqueue);
	}
	j1939_set_event(J1939_RX_TASK_READY);

	while (1)
	{
		xNextExpireTime = portMAX_DELAY;
		complete_frame = NO_NEED_PORCESS_COMPLETE_FRAME;
		for (i = 0; i < rx_session_num; i++)
		{
			uint8_t timer_event = 0;
			if (rx_session[i].runtime[0].state != J1939TP_RX_IDLE && rx_session[i].max_len != 0)
			{
				/*DA != 0xFF,we should do below things*/
				if (rx_session[i].runtime[0].timer.valid)
				{
					if (xTaskCheckTickTimeout(rx_session[i].runtime[0].timer.end_tick))
					{
						timer_event = rx_session[i].runtime[0].timer.timeout_type;
						if (timer_event == SEND_CTS_EVENT)
						{
							j1939tp_send_CTS_message(j1939tp_runtime, rx_session);
							j1939tp_set_timer(&rx_session->runtime[0].timer, config->timeout_T2, T2_TIMEOUT);
							rx_session->runtime[0].state = J1939TP_RX_RECEIVING_DT;
						}
						else if (timer_event == T1_TIMEOUT || timer_event == T2_TIMEOUT)
						{
							uint8_t sa = rx_session[i].runtime[0].saddr;
							uint8_t da = rx_session[i].addr;
							uint32_t pgn = rx_session[i].runtime[0].pgn;
							uint8_t priority = rx_session[i].runtime[0].priority;
							rx_session[i].runtime[0].state = J1939TP_RX_IDLE;
							j1939tp_send_ConnectionAbort_message(j1939tp_runtime, sa, da, pgn, priority,
																 J1939_TP_ABORT_TIMEOUT);
							log_d("j1939tp:%s,%s tiemout\n", j1939_tp_abort_to_str(J1939_TP_ABORT_TIMEOUT),
								  timer_event == T1_TIMEOUT ? "T1" : "T2");
							j1939tp_close_rx_session(&rx_session[i], SPECIFIC_SESSION);
						}
						else if (timer_event == SEND_EOMA_EVENT)
						{
							j1939tp_send_EOMA_message(j1939tp_runtime, rx_session);
							j1939tp_close_rx_session(&rx_session[i], SPECIFIC_SESSION);
						}
						else
						{
							j1939tp_close_rx_session(&rx_session[i], SPECIFIC_SESSION);
						}
					}
				}
			}
			if (rx_session[i].runtime[1].state != J1939TP_RX_IDLE && rx_session[i].max_bam_len != 0)
			{
				/*DA == 0xFF,we should do below things*/
				if (rx_session[i].runtime[1].timer.valid)
				{
					if (xTaskCheckTickTimeout(rx_session[i].runtime[1].timer.end_tick))
					{
						timer_event = rx_session[i].runtime[1].timer.timeout_type;
						if (timer_event == T1_TIMEOUT)
						{
							log_d("j1939tp:%s,broadcast T1 tiemout\n", j1939_tp_abort_to_str(J1939_TP_ABORT_TIMEOUT));
						}
						j1939tp_close_rx_session(&rx_session[i], GLOBAL_SESSION);
					}
				}
			}
		}
		if (ret == pdPASS) /*parse new CAN frame*/
		{
			// log_d("j1939 recv new can frame.");
			j1939_msg_info.addr.pgn = (element.can_id >> 8) & J1939_PGN_MAX;
			j1939_msg_info.addr.sa = GET_J1939_SOURCE_ADDR(element.can_id);
			j1939_msg_info.priority = GET_J1939_PRIORITY(element.can_id);
			if (j1939_pgn_is_pdu1(j1939_msg_info.addr.pgn))
			{
				j1939_msg_info.addr.da = j1939_msg_info.addr.pgn & 0xff;
				j1939_msg_info.addr.pgn &= J1939_PGN_PDU1_MAX;
			}
			else
			{
				j1939_msg_info.addr.da = J1939_NO_ADDR;
			}
			j1939_msg_info.addr.type = J1939_TP;
			complete_frame = j1939tp_recv(j1939tp_runtime, &j1939_msg_info, &element);
		}
        
		if (complete_frame >= 0)
		{
			int pgn_index = 0;
			if (j1939_msg_info.addr.type == J1939_SIMPLE)
			{
                pgn_index = pgn_bin_search(config->pgn_tab, j1939_msg_info.addr.pgn, config->pgn_num);
				if (pgn_index >= 0)
				{
                    // log_d("recv pgn_index %d", pgn_index);
					if (config->pgn_tab[pgn_index].handle_func)
						config->pgn_tab[pgn_index].handle_func(&j1939_msg_info, element.data, 8);			/* 返回实际的帧载荷 */
				}
			}
			else
			{
				uint32_t pgn = 0;
				uint8_t *data = NULL;
				uint32_t len = 0;
				j1939_msg_info_t info;
				if (j1939_address_is_unicast(j1939_msg_info.addr.da))
				{
					/* 多帧点对点 */
					pgn = rx_session[complete_frame].runtime[0].pgn;
					data = rx_session[complete_frame].buffer;
					len = rx_session[complete_frame].runtime[0].total_msg_size;
					info.priority = rx_session[complete_frame].runtime[0].priority;
					info.addr.pgn = rx_session[complete_frame].runtime[0].pgn;
					info.addr.sa = rx_session[complete_frame].addr;
					info.addr.da = rx_session[complete_frame].runtime[0].saddr;
					/*after send EOMA,close rx session*/
					//j1939tp_close_rx_session(&rx_session[complete_frame],SPECIFIC_SESSION);
				}
				else
				{
					/* 多帧广播 */
					pgn = rx_session[complete_frame].runtime[1].pgn;
					data = (rx_session[complete_frame].buffer + rx_session[complete_frame].max_len);
					len = rx_session[complete_frame].runtime[1].total_msg_size;
					info.priority = rx_session[complete_frame].runtime[1].priority;
					info.addr.pgn = rx_session[complete_frame].runtime[1].pgn;
					info.addr.sa = rx_session[complete_frame].addr;
					info.addr.da = J1939_BROADCAST_ADDR;
					j1939tp_close_rx_session(&rx_session[complete_frame], GLOBAL_SESSION);
				}
				pgn_index = pgn_bin_search(config->pgn_tab, pgn, config->pgn_num);
				if (pgn_index >= 0)
				{
					if (config->pgn_tab[pgn_index].handle_func)
						config->pgn_tab[pgn_index].handle_func(&info, data, len);
				}
			}
		}

		for (i = 0; i < rx_session_num; i++)
		{
			if (rx_session[i].runtime[0].timer.valid)
			{
				if (xNextExpireTime < rx_session[i].runtime[0].timer.end_tick)
					xNextExpireTime = rx_session[i].runtime[0].timer.end_tick;
			}

			if (rx_session[i].runtime[1].timer.valid)
			{
				if (xNextExpireTime < rx_session[i].runtime[1].timer.end_tick)
					xNextExpireTime = rx_session[i].runtime[1].timer.end_tick;
			}
		}
		xTimeNow = xTaskGetTickCount();
		if (xNextExpireTime > xTimeNow)
		{
			if (xNextExpireTime == portMAX_DELAY)
				ret = xQueueReceive(config->rxqueue, &element, pdMS_TO_TICKS(100));
			else
				ret = xQueueReceive(config->rxqueue, &element, (xNextExpireTime - xTimeNow));
		}
		else
			ret = xQueueReceive(config->rxqueue, &element, 5u);
	}
}

int j1939tp_transmit(j1939tp_runtime_t *runtime, j1939_msg_info_t *info, uint8_t *buffer, uint16_t length)
{
	j1939tp_tx_sesssion_t *tx_session = runtime->session->tx;
	int ret = 0;
	int index = 0;
	if (length <= 8)
	{
		struct can_frame element = {0};
		memset(&element, 0, sizeof(struct can_frame));
		memset(element.data, runtime->config->padding, 8);
		element.can_id = j1939_get_canid(info->addr.sa, info->addr.da, info->addr.pgn, info->priority);
		// log_d("length <= 8 can_id %04x", element.can_id);
		element.len = 8;
		memcpy(element.data, buffer, length);
		j1939tp_write_by_channel(runtime->config->chn, &element);
	}
	else
	{
		index = tx_ecu_bin_search(runtime->session->tx, info->addr.sa, runtime->session->tx_num);
		if (index >= 0)
		{
			j1939_output_msg_info_t msg = {0};
			msg.msg_code = J1939_MULTIFRAME_MSG;
			if (info->addr.da == J1939_BROADCAST_ADDR)
			{
				if (tx_session[index].runtime[1].state == J1939TP_TX_IDLE && length <= tx_session[index].max_bam_len)
				{
					tx_session[index].runtime[1].state = J1939TP_TX_LOCK_BUFFER;
					tx_session[index].runtime[1].pgn = info->addr.pgn;
					tx_session[index].runtime[1].priority = info->priority;
					tx_session[index].runtime[1].daddr = info->addr.da;
					tx_session[index].runtime[1].total_msg_size = length;
					tx_session[index].runtime[1].tx_total_packets = 0;
					tx_session[index].runtime[1].waitForCTS_packets = 0;
					tx_session[index].runtime[1].tx_max_packets = 0;
					tx_session[index].runtime[1].retransmit_cnt = 0;
					memcpy((tx_session[index].buffer + tx_session[index].max_len), buffer, length);
					msg.index = (uint8_t)index;
					msg.type = J1939_TX_GLOBAL_MSG;
					ret = xQueueSend(runtime->config->txqueue, &msg, 0);
				}
				else
					ret = -1;
			}
			else
			{
				if (tx_session[index].runtime[0].state == J1939TP_TX_IDLE && length <= tx_session[index].max_len)
				{
					tx_session[index].runtime[0].state = J1939TP_TX_LOCK_BUFFER;
					tx_session[index].runtime[0].pgn = info->addr.pgn;
					tx_session[index].runtime[0].priority = info->priority;
					tx_session[index].runtime[0].daddr = info->addr.da;
					tx_session[index].runtime[0].total_msg_size = length;
					tx_session[index].runtime[0].tx_total_packets = 0;
					tx_session[index].runtime[0].waitForCTS_packets = 0;
					tx_session[index].runtime[0].tx_max_packets =
						(runtime->config->tp_block) ? (runtime->config->tp_block) : 255;
					tx_session[index].runtime[0].retransmit_cnt = 0;
					memcpy(tx_session[index].buffer, buffer, length);
					msg.index = (uint8_t)index;
					msg.type = J1939_TX_SPECIFIC_MSG;
					xQueueSend(runtime->config->txqueue, &msg, 0);
				}
				else
					ret = -1;
			}
		}
		else
			ret = -1;
	}

	return ret;
}

void j1939tp_tx_cm(j1939tp_runtime_t *runtime, struct can_frame *element)
{
	uint8_t cmd = element->data[0];
	int index = 0;
	uint8_t da = 0;
	pgn_t pgn = 0;
	pgn = (element->can_id >> 8) & J1939_PGN_MAX;
	if (j1939_pgn_is_pdu1(pgn))
	{
		da = pgn & 0xff;
	}
	else
	{
		da = J1939_BROADCAST_ADDR;
	}
	index = tx_ecu_bin_search(runtime->session->tx, da, runtime->session->tx_num);
	switch (cmd)
	{
	case J1939_TP_CMD_CTS:
		if (index >= 0)
		{
			pgn = (uint32_t)element->data[5] | (uint32_t)element->data[6] << 8 | (uint32_t)element->data[7] << 16;
			j1939tp_tx_sesssion_t *tx_session = &runtime->session->tx[index];
			if (pgn == tx_session->runtime[0].pgn)
			{
				if (tx_session->runtime[0].state == J1939TP_TX_WAIT_FOR_CTS)
				{
					tx_session->runtime[0].waitForCTS_packets = element->data[1];	/* 接收方可以接收的数据包数 */
					if (tx_session->runtime[0].waitForCTS_packets == HOLD_CONNECTION_OPEN_NUM)
					{
						tx_session->runtime[0].state = J1939TP_TX_WAIT_FOR_CTS;
						j1939tp_set_timer(&tx_session->runtime[0].timer, runtime->config->timeout_T4, T4_TIMEOUT);
					}
					else
					{
						if (element->data[2])
						{
							if (element->data[2] <= tx_session->runtime[0].tx_total_packets)
							{
								if (tx_session->runtime[0].retransmit_cnt < RETRANSMIT_NO_LIMIT)
									tx_session->runtime[0].retransmit_cnt++;
							}
							tx_session->runtime[0].tx_total_packets = element->data[2] - 1;
							tx_session->runtime[0].state = J1939TP_TX_SENDING_DT;
							j1939tp_set_timer(&tx_session->runtime[0].timer, runtime->config->timeout_Tr,
											  SEND_DT_EVENT);
						}
						else
						{
							uint8_t da = tx_session->runtime[0].daddr;
							uint8_t sa = tx_session->addr;
							uint8_t priority = tx_session->runtime[0].priority;
							j1939tp_send_ConnectionAbort_message(runtime, sa, da, pgn, priority,
																 J1939_TP_ABORT_BAD_SEQ);
							log_d("j1939tp:%s\n", j1939_tp_abort_to_str(J1939_TP_ABORT_BAD_SEQ));
							j1939tp_close_tx_session(tx_session, SPECIFIC_SESSION);
						}
					}
				}
				else if (tx_session->runtime[0].state == J1939TP_TX_SENDING_DT)
				{
					uint8_t da = tx_session->runtime[0].daddr;
					uint8_t sa = tx_session->addr;
					uint8_t priority = tx_session->runtime[0].priority;
					j1939tp_send_ConnectionAbort_message(runtime, sa, da, pgn, priority, J1939_TP_ABORT_GENERIC);
					log_d("j1939tp:%s\n", j1939_tp_abort_to_str(J1939_TP_ABORT_GENERIC));
					j1939tp_close_tx_session(tx_session, SPECIFIC_SESSION);
				}
			}
		}
		break;
	case J1939_TP_CMD_EOMA:
		if (index >= 0)
		{
			pgn = (uint32_t)element->data[5] | (uint32_t)element->data[6] << 8 | (uint32_t)element->data[7] << 16;
			j1939tp_tx_sesssion_t *tx_session = &runtime->session->tx[index];
			if (pgn == tx_session->runtime[0].pgn && tx_session->runtime[0].state == J1939TP_TX_WAIT_FOR_ENDOFMSGACK)
			{
				tx_session->runtime[0].state = J1939TP_TX_IDLE;
				j1939tp_close_tx_session(tx_session, SPECIFIC_SESSION);
			}
		}
		break;
	case J1939_TP_CMD_ABORT:
		if (index >= 0)
		{
			pgn = (uint32_t)element->data[5] | (uint32_t)element->data[6] << 8 | (uint32_t)element->data[7] << 16;
			j1939tp_tx_sesssion_t *tx_session = &runtime->session->tx[index];
			if (pgn == tx_session->runtime[0].pgn)
			{
				tx_session->runtime[0].state = J1939TP_TX_IDLE;
				j1939tp_close_tx_session(tx_session, SPECIFIC_SESSION);
			}
		}
		break;
	}
}

/* j1939发送端请求发送数据 */
static void j1939tp_send_RTS_message(j1939tp_runtime_t *runtime, j1939tp_tx_sesssion_t *session)
{
	struct can_frame element;
	memset(&element, 0, sizeof(struct can_frame));
	memset(element.data, runtime->config->padding, 8);
	element.can_id =
		j1939_get_canid(session->addr, session->runtime[0].daddr, J1939TP_CM_PGN, session->runtime[0].priority);
	element.len = 8;
	element.data[0] = J1939_TP_CMD_RTS;

	session->runtime[0].tx_total_packets = 0;
	element.data[1] = session->runtime[0].total_msg_size & 0xFF;
	element.data[2] = (session->runtime[0].total_msg_size >> 8) & 0xFF;
	element.data[3] = SIZE2PACKETS(session->runtime[0].total_msg_size);
	element.data[4] = (runtime->config->tp_block) ? (runtime->config->tp_block) : 255;

	element.data[5] = session->runtime[0].pgn & 0xFF;
	element.data[6] = (session->runtime[0].pgn & 0xFF00) >> 8;
	element.data[7] = (session->runtime[0].pgn & 0xFF0000) >> 16;

	j1939tp_write_by_channel(runtime->config->chn, &element);
}

static void j1939tp_send_BAM_message(j1939tp_runtime_t *runtime, j1939tp_tx_sesssion_t *session)
{
	struct can_frame element;
	memset(&element, 0, sizeof(struct can_frame));
	memset(element.data, runtime->config->padding, 8);
	element.can_id = j1939_get_canid(session->addr, J1939_BROADCAST_ADDR, J1939TP_CM_PGN, session->runtime[1].priority);
	element.len = 8;
	element.data[0] = J1939_TP_CMD_BAM;									/* 广播报文的命令类型0x20 */

	session->runtime[1].tx_total_packets = 0;
	element.data[1] = session->runtime[1].total_msg_size & 0xFF;		/* 总的数据字节数L */
	element.data[2] = (session->runtime[1].total_msg_size >> 8) & 0xFF;	/* 总的数据字节数H */
	element.data[3] = SIZE2PACKETS(session->runtime[1].total_msg_size);	/* 总的发送包数 */

	element.data[5] = session->runtime[1].pgn & 0xFF;					/* 参数组3byte */
	element.data[6] = (session->runtime[1].pgn & 0xFF00) >> 8;
	element.data[7] = (session->runtime[1].pgn & 0xFF0000) >> 16;

	j1939tp_write_by_channel(runtime->config->chn, &element);			/* 链路层发送出去 */
}

static void j1939tp_send_DT_message(j1939tp_runtime_t *runtime, j1939tp_tx_sesssion_t *session, uint8_t type)
{
	struct can_frame element;
	memset(&element, 0, sizeof(struct can_frame));
	memset(element.data, runtime->config->padding, 8);
	if (type <= GLOBAL_SESSION)
	{
		uint8_t da = (type == GLOBAL_SESSION) ? J1939_BROADCAST_ADDR : session->runtime[type].daddr;
		uint32_t pos = 0;
		uint32_t len = 0;
		element.can_id = j1939_get_canid(session->addr, da, J1939TP_DT_PGN, session->runtime[type].priority);
		element.len = 8;
		element.data[0] = session->runtime[type].tx_total_packets + 1;	/* 数据包的编号 */
		pos = session->runtime[type].tx_total_packets * 7;
		pos += ((type == GLOBAL_SESSION) ? session->max_len : 0);
		len = session->runtime[type].total_msg_size - session->runtime[type].tx_total_packets * 7;
		if (len >= 7)
			memcpy(&element.data[1], (session->buffer + pos), 7);
		else
			memcpy(&element.data[1], (session->buffer + pos), len);
		j1939tp_write_by_channel(runtime->config->chn, &element);
	}
}

void j1939tp_send(j1939tp_runtime_t *runtime, j1939_output_msg_info_t *msg_info)
{
	uint8_t index = 0;
	switch (msg_info->msg_code)
	{
	case J1939_CTS_MSG:
		j1939tp_tx_cm(runtime, &msg_info->frame);
		break;
	case J1939_EOMA_MSG:
		j1939tp_tx_cm(runtime, &msg_info->frame);
		break;
	case J1939_ABORT_MSG:
		j1939tp_tx_cm(runtime, &msg_info->frame);
		break;
	/* BAM广播方式传送多条报文 */
	case J1939_MULTIFRAME_MSG:
		index = msg_info->index;
		if (msg_info->type == J1939_TX_SPECIFIC_MSG)
		{
			/* 点对点的方式 */
			j1939tp_send_RTS_message(runtime, &runtime->session->tx[index]);			/* 发送方，传输数据请求 */
			runtime->session->tx[index].runtime[0].state = J1939TP_TX_WAIT_FOR_CTS;
			j1939tp_set_timer(&runtime->session->tx[index].runtime[0].timer, runtime->config->timeout_T3, T3_TIMEOUT);
		}
		else
		{
			/* 广播方式 */
			j1939tp_send_BAM_message(runtime, &runtime->session->tx[index]);
			runtime->session->tx[index].runtime[1].state = J1939TP_TX_SENDING_DT;
			j1939tp_set_timer(&runtime->session->tx[index].runtime[1].timer, runtime->config->timeout_Tr,
							  SEND_DT_EVENT);
		}
		break;
	}
}

void j1939tp_tx_task(void *arg)
{
	j1939tp_runtime_t *j1939tp_runtime = (j1939tp_runtime_t *)arg;
	const j1939tp_config_t *config = j1939tp_runtime->config;
	j1939tp_tx_sesssion_t *tx_session = j1939tp_runtime->session->tx;
	uint16_t tx_session_num = j1939tp_runtime->session->tx_num;
	TickType_t xNextExpireTime = 0;
	TickType_t xTimeNow = 0;
	j1939_output_msg_info_t msg = {0};
	int ret = 0;
	uint16_t i = 0;

	if (config->txqueue)
	{
		xQueueCreateStatic(config->tx_queue_size, sizeof(j1939_output_msg_info_t), config->txq_storage,
						   config->txqueue);
	}
	j1939_set_event(J1939_TX_TASK_READY);

	while (1)
	{
		xNextExpireTime = portMAX_DELAY;
		for (i = 0; i < tx_session_num; i++)
		{
			uint8_t timer_event = 0;
			if (tx_session[i].runtime[0].state > J1939TP_TX_LOCK_BUFFER && tx_session[i].max_len != 0)
			{
				if (tx_session[i].runtime[0].timer.valid)
				{
					if (xTaskCheckTickTimeout(tx_session[i].runtime[0].timer.end_tick))
					{
						timer_event = tx_session[i].runtime[0].timer.timeout_type;
						if (timer_event == SEND_DT_EVENT)
						{
							j1939tp_send_DT_message(j1939tp_runtime, &tx_session[i], SPECIFIC_SESSION);
							tx_session[i].runtime[0].tx_total_packets++;
							tx_session[i].runtime[0].waitForCTS_packets--;
							if (tx_session[i].runtime[0].tx_total_packets >=
								SIZE2PACKETS(tx_session[i].runtime[0].total_msg_size))
							{
								tx_session[i].runtime[0].state = J1939TP_TX_WAIT_FOR_ENDOFMSGACK;
								j1939tp_set_timer(&tx_session[i].runtime[0].timer, config->timeout_T3, T3_TIMEOUT);
							}
							else if (tx_session[i].runtime[0].waitForCTS_packets == 0)
							{
								tx_session[i].runtime[0].state = J1939TP_TX_WAIT_FOR_CTS;
								j1939tp_set_timer(&tx_session[i].runtime[0].timer, config->timeout_T3, T3_TIMEOUT);
							}
							else if (tx_session[i].runtime[0].retransmit_cnt > config->retransmit_max &&
									 tx_session[i].runtime[0].retransmit_cnt != RETRANSMIT_NO_LIMIT)
							{
								uint8_t da = tx_session[i].runtime[0].daddr;
								uint8_t sa = tx_session[i].addr;
								uint32_t pgn = tx_session[i].runtime[0].pgn;
								uint8_t priority = tx_session[i].runtime[0].priority;
								j1939tp_send_ConnectionAbort_message(j1939tp_runtime, sa, da, pgn, priority,
																	 J1939_TP_ABORT_FAULT);
								log_d("j1939tp:%s\n", j1939_tp_abort_to_str(J1939_TP_ABORT_FAULT));
								j1939tp_close_tx_session(&tx_session[i], SPECIFIC_SESSION);
							}
							else
								j1939tp_set_timer(&tx_session[i].runtime[0].timer, config->timeout_Tr, SEND_DT_EVENT);
						}
						else if (timer_event == T3_TIMEOUT || timer_event == T4_TIMEOUT)
						{
							uint8_t da = tx_session[i].runtime[0].daddr;
							uint8_t sa = tx_session[i].addr;
							uint32_t pgn = tx_session[i].runtime[0].pgn;
							uint8_t priority = tx_session[i].runtime[0].priority;
							j1939tp_send_ConnectionAbort_message(j1939tp_runtime, sa, da, pgn, priority,
																 J1939_TP_ABORT_TIMEOUT);
							log_d("j1939tp:%s,%s tiemout\n", j1939_tp_abort_to_str(J1939_TP_ABORT_TIMEOUT),
								  timer_event == T3_TIMEOUT ? "T3" : "T4");
							j1939tp_close_tx_session(&tx_session[i], SPECIFIC_SESSION);
						}
						else
							j1939tp_close_tx_session(&tx_session[i], SPECIFIC_SESSION);
					}
				}
			}

			/* 广播方式发送数据，超时判断根据Time_Tr时间发送 */
			if (tx_session[i].runtime[1].state > J1939TP_TX_LOCK_BUFFER && tx_session[i].max_bam_len != 0)
			{
				if (xTaskCheckTickTimeout(tx_session[i].runtime[1].timer.end_tick))
				{
					timer_event = tx_session[i].runtime[1].timer.timeout_type;
					if (timer_event == SEND_DT_EVENT)
					{
						j1939tp_send_DT_message(j1939tp_runtime, &tx_session[i], GLOBAL_SESSION);
						tx_session[i].runtime[1].tx_total_packets++;
						/* 所有数据包发送完成 */
						if (tx_session[i].runtime[1].tx_total_packets >=
							SIZE2PACKETS(tx_session[i].runtime[1].total_msg_size))
						{
							tx_session[i].runtime[1].state = J1939TP_TX_IDLE;
							j1939tp_close_tx_session(&tx_session[i], GLOBAL_SESSION);
						}
						else
							j1939tp_set_timer(&tx_session[i].runtime[1].timer, config->timeout_Tr, SEND_DT_EVENT);
					}
					else
						j1939tp_close_tx_session(&tx_session[i], GLOBAL_SESSION);
				}
			}
		}

		if (ret == pdPASS) /*parse new notification*/
		{
			j1939tp_send(j1939tp_runtime, &msg);
		}
		for (i = 0; i < tx_session_num; i++)
		{
			if (tx_session[i].runtime[0].timer.valid)
			{
				if (xNextExpireTime < tx_session[i].runtime[0].timer.end_tick)
					xNextExpireTime = tx_session[i].runtime[0].timer.end_tick;
			}

			if (tx_session[i].runtime[1].timer.valid)
			{
				if (xNextExpireTime < tx_session[i].runtime[1].timer.end_tick)
					xNextExpireTime = tx_session[i].runtime[1].timer.end_tick;
			}
		}

		xTimeNow = xTaskGetTickCount();
		if (xNextExpireTime > xTimeNow)
		{
			if (xNextExpireTime == portMAX_DELAY)
				ret = xQueueReceive(config->txqueue, &msg, pdMS_TO_TICKS(100));
			else
				ret = xQueueReceive(config->txqueue, &msg, (xNextExpireTime - xTimeNow));
		}
		else
			ret = xQueueReceive(config->txqueue, &msg, 5u);
	}
}
